//! 集群集成测试

use nacos_cluster::{ClusterConfig, ClusterManager, ClusterNodeInfo, ClusterState};
use nacos_core::NodeState;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

/// 测试多节点集群
#[tokio::test]
async fn test_multi_node_cluster() {
    let (tx, mut rx) = mpsc::unbounded_channel();
    
    // 创建3个节点的集群
    let config1 = ClusterConfig {
        cluster_name: "test-cluster".to_string(),
        current_node_id: "node1".to_string(),
        heartbeat_timeout: 2,
        ..Default::default()
    };
    
    let config2 = ClusterConfig {
        cluster_name: "test-cluster".to_string(),
        current_node_id: "node2".to_string(),
        heartbeat_timeout: 2,
        ..Default::default()
    };
    
    let config3 = ClusterConfig {
        cluster_name: "test-cluster".to_string(),
        current_node_id: "node3".to_string(),
        heartbeat_timeout: 2,
        ..Default::default()
    };
    
    let manager1 = Arc::new(ClusterManager::new(config1));
    let manager2 = Arc::new(ClusterManager::new(config2));
    let manager3 = Arc::new(ClusterManager::new(config3));
    
    manager1.set_event_sender(tx.clone());
    manager2.set_event_sender(tx.clone());
    manager3.set_event_sender(tx.clone());
    
    // 创建节点
    let node1 = ClusterNodeInfo::new("127.0.0.1", 8848, 3);
    let node2 = ClusterNodeInfo::new("127.0.0.1", 8849, 2);
    let node3 = ClusterNodeInfo::new("127.0.0.1", 8850, 1);
    
    // 初始化集群
    manager1.initialize(node1.clone()).await.unwrap();
    manager1.add_node(node2.clone()).await.unwrap();
    manager1.add_node(node3.clone()).await.unwrap();
    
    // 设置领导者（权重最高的node1）
    manager1.set_leader(&node1.node_id).await.unwrap();
    
    // 验证集群状态
    assert_eq!(manager1.get_cluster_size().await, 3);
    assert_eq!(manager1.get_healthy_nodes().await.len(), 3);
    assert!(manager1.get_leader().await.is_some());
    assert_eq!(manager1.get_leader().await.unwrap().node_id, node1.node_id);
    
    // 等待事件处理
    sleep(Duration::from_millis(100)).await;
    
    // 模拟节点故障
    manager1.update_node_state(&node2.node_id, NodeState::Failed).await.unwrap();
    
    // 验证集群状态更新
    assert_eq!(manager1.get_healthy_nodes().await.len(), 2);
    
    // 领导者应该仍然是node1，因为它还健康
    assert_eq!(manager1.get_leader().await.unwrap().node_id, node1.node_id);
    
    // 模拟领导者故障，应该触发重新选举
    manager1.update_node_state(&node1.node_id, NodeState::Failed).await.unwrap();
    
    // 新的领导者应该是node3（剩下的节点中权重最高的）
    let new_leader = manager1.get_leader().await.unwrap();
    assert!(new_leader.node_id == node3.node_id);
}

/// 测试网络分区恢复
#[tokio::test]
async fn test_network_partition_recovery() {
    let (tx, mut rx) = mpsc::unbounded_channel();
    
    let config = ClusterConfig {
        cluster_name: "partition-test".to_string(),
        heartbeat_timeout: 1,
        ..Default::default()
    };
    
    let manager = Arc::new(ClusterManager::new(config));
    manager.set_event_sender(tx);
    
    // 创建5个节点的集群
    let nodes = vec![
        ClusterNodeInfo::new("127.0.0.1", 8848, 5),
        ClusterNodeInfo::new("127.0.0.1", 8849, 4),
        ClusterNodeInfo::new("127.0.0.1", 8850, 3),
        ClusterNodeInfo::new("127.0.0.1", 8851, 2),
        ClusterNodeInfo::new("127.0.0.1", 8852, 1),
    ];
    
    // 初始化第一个节点
    manager.initialize(nodes[0].clone()).await.unwrap();
    
    // 添加其他节点
    for node in &nodes[1..] {
        manager.add_node(node.clone()).await.unwrap();
    }
    
    // 设置初始领导者
    manager.set_leader(&nodes[0].node_id).await.unwrap();
    
    // 初始状态应该是Running
    assert_eq!(manager.get_cluster_state().await, ClusterState::Running);
    
    // 模拟网络分区：3个节点故障
    manager.update_node_state(&nodes[2].node_id, NodeState::Failed).await.unwrap();
    manager.update_node_state(&nodes[3].node_id, NodeState::Failed).await.unwrap();
    manager.update_node_state(&nodes[4].node_id, NodeState::Failed).await.unwrap();
    
    // 集群应该进入降级模式
    assert_eq!(manager.get_cluster_state(), ClusterState::Degraded);
    
    // 恢复网络分区
    manager.update_node_state(&nodes[2].node_id, NodeState::Running).await.unwrap();
    manager.update_node_state(&nodes[3].node_id, NodeState::Running).await.unwrap();
    manager.update_node_state(&nodes[4].node_id, NodeState::Running).await.unwrap();
    
    // 集群应该恢复Running状态
    assert_eq!(manager.get_cluster_state().await, ClusterState::Running);
}

/// 测试自动心跳检测
#[tokio::test]
async fn test_automatic_heartbeat_detection() {
    let (tx, mut rx) = mpsc::unbounded_channel();
    
    let config = ClusterConfig {
        cluster_name: "heartbeat-test".to_string(),
        heartbeat_timeout: 1,
        ..Default::default()
    };
    
    let manager = Arc::new(ClusterManager::new(config));
    manager.set_event_sender(tx);
    
    let node = ClusterNodeInfo::new("127.0.0.1", 8848, 1);
    manager.initialize(node.clone()).await.unwrap();
    
    // 启动心跳检查
    let manager_clone = manager.clone();
    manager.start_heartbeat_check().await;
    
    // 模拟心跳超时
    sleep(Duration::from_secs(2)).await;
    
    // 节点应该被标记为故障
    let stats = manager.get_cluster_stats().await;
    assert_eq!(stats.healthy_nodes, 0);
    
    // 更新心跳
    manager.update_node_heartbeat(&node.node_id).await.unwrap();
    
    // 节点应该恢复为健康
    let stats = manager.get_cluster_stats().await;
    assert_eq!(stats.healthy_nodes, 1);
}

/// 测试权重选举
#[tokio::test]
async fn test_weight_based_election() {
    let (tx, _rx) = mpsc::unbounded_channel();
    
    let config = ClusterConfig {
        cluster_name: "weight-test".to_string(),
        ..Default::default()
    };
    
    let manager = Arc::new(ClusterManager::new(config));
    manager.set_event_sender(tx);
    
    // 创建不同权重的节点
    let nodes = vec![
        ClusterNodeInfo::new("127.0.0.1", 8848, 1),
        ClusterNodeInfo::new("127.0.0.1", 8849, 5), // 最高权重
        ClusterNodeInfo::new("127.0.0.1", 8850, 3),
        ClusterNodeInfo::new("127.0.0.1", 8851, 2),
    ];
    
    manager.initialize(nodes[0].clone()).await.unwrap();
    
    for node in &nodes[1..] {
        manager.add_node(node.clone()).await.unwrap();
    }
    
    // 触发选举，权重最高的应该成为领导者
    manager.trigger_election().await.unwrap();
    
    let leader = manager.get_leader().unwrap();
    assert_eq!(leader.node_id, nodes[1].node_id); // 权重为5的节点
    assert_eq!(leader.weight, 5);
}

/// 测试并发节点操作
#[tokio::test]
async fn test_concurrent_node_operations() {
    use std::sync::Arc;
    use tokio::task;
    
    let (tx, _rx) = mpsc::unbounded_channel();
    
    let config = ClusterConfig {
        cluster_name: "concurrent-test".to_string(),
        ..Default::default()
    };
    
    let manager = Arc::new(ClusterManager::new(config));
    manager.set_event_sender(tx);
    
    let node = ClusterNodeInfo::new("127.0.0.1", 8848, 1);
    manager.initialize(node.clone()).await.unwrap();
    
    // 并发添加多个节点
    let mut handles = vec![];
    for i in 0..10 {
        let manager_clone = manager.clone();
        let handle = task::spawn(async move {
            let new_node = ClusterNodeInfo::new("127.0.0.1", 8850 + i, 1);
            manager_clone.add_node(new_node).await
        });
        handles.push(handle);
    }
    
    // 等待所有操作完成
    for handle in handles {
        handle.await.unwrap().unwrap();
    }
    
    // 应该有11个节点（初始节点 + 10个新节点）
    assert_eq!(manager.get_cluster_size(), 11);
}

/// 测试集群统计信息
#[tokio::test]
async fn test_cluster_statistics() {
    let (tx, _rx) = mpsc::unbounded_channel();
    
    let config = ClusterConfig {
        cluster_name: "stats-test".to_string(),
        ..Default::default()
    };
    
    let manager = Arc::new(ClusterManager::new(config));
    manager.set_event_sender(tx);
    
    // 创建多个节点
    let nodes = vec![
        ClusterNodeInfo::new("127.0.0.1", 8848, 1),
        ClusterNodeInfo::new("127.0.0.1", 8849, 1),
        ClusterNodeInfo::new("127.0.0.1", 8850, 1),
    ];
    
    manager.initialize(nodes[0].clone()).await.unwrap();
    
    for node in &nodes[1..] {
        manager.add_node(node.clone()).await.unwrap();
    }
    
    // 设置领导者
    manager.set_leader(&nodes[0].node_id).await.unwrap();
    
    let stats = manager.get_cluster_stats().await;
    assert_eq!(stats.total_nodes, 3);
    assert_eq!(stats.healthy_nodes, 3);
    assert!(stats.leader_id.is_some());
    assert_eq!(stats.cluster_state, ClusterState::Running);
}